package com.calabar.phm.etl.operator.output;

import com.calabar.phm.etl.operator.api.ITransformer;
import com.calabar.phm.etl.operator.api.OutputTransformer;
import com.google.gson.Gson;
import com.google.gson.JsonIOException;
import com.google.gson.JsonSyntaxException;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.sql.BatchUpdateException;
import java.util.Map;
import java.util.Properties;

/**
 * @Author duanyuping <yuping.duan@cdcalabar.com>
 * @Date 17-7-19 上午9:37
 * @Version v1.0
 * @Des oracle输出算子
 */
public class OracleWriter extends ITransformer implements OutputTransformer {

    private static final Logger LOG = Logger.getLogger(OracleWriter.class);
    /**
     * json参数
     */
    private  OracleParams params;

    public OracleWriter(String uid, String paramsJson) throws Exception {
        super(uid, paramsJson);
    }


    /**
     * 解析与参数校验
     * @throws Exception
     */
    @Override
    public void validateParams() throws Exception {
        if(StringUtils.isBlank(getParamsJson())){
            throw  new Exception("Oracle数据库输出算子参数为空!");
        }

        //解析校验json
        Gson gson = new Gson();
        try{
            params = gson.fromJson(getParamsJson(), OracleParams.class);

        }catch (JsonIOException e){
            throw  new JsonIOException("Oracle数据库输出算子参数出现传输错误!", e);
        }catch (JsonSyntaxException e){
            throw  new JsonSyntaxException("Oracle数据库输出算子参数语法错误!", e);
        }catch (Exception e){
            throw  new Exception("Oracle数据库输出算子参数解析失败!", e);
        }

        //参数校验
        if(StringUtils.isBlank(params.getUrl())) {
            throw  new Exception("Oracle数据库输出算子路径不能为空!");
        }

        if(StringUtils.isBlank(params.getUser())) {
            throw  new Exception("Oracle数据库输出算子user不能为空!");
        }

        if(StringUtils.isBlank(params.getPassword())) {
            throw  new Exception("Oracle数据库输出算子password不能为空!");
        }

        if(StringUtils.isBlank(params.getTable())) {
            throw  new Exception("Oracle数据库输出算子table不能为空!");
        }

        if(MapUtils.isEmpty(params.getDataType())) {
            throw  new Exception("Oracle数据库输出算子Data_type不能为空!");
        }



    }


    public void transform(SparkSession session, Dataset<Row> df) throws Exception {

        //更改数据类型
        try{
            Map<String, String> schemaMap = params.getDataType();
            for (Map.Entry<String, String> entry: schemaMap.entrySet()) {
                df = df.withColumn(entry.getKey(), df.col(entry.getKey()).cast(entry.getValue()));

            }
        }catch (Exception e){
            throw  new Exception("Oracle数据库输出算子数据转换失败!");
        }


        //保存数据
        String bigURL = params.getUrl();
        String table = params.getTable();
        try{
            Properties connectionProperties = new Properties();
            connectionProperties.put("user", params.getUser());
            connectionProperties.put("password", params.getPassword());
            connectionProperties.put("driver", "oracle.jdbc.driver.OracleDriver");
            connectionProperties.put("truncate", "true");
            connectionProperties.put("batchsize", "500000");


            if (params.getSavemodel() == OracleWriterEnum.APPEND.getIndexs()) {

                df.write().mode(SaveMode.Append).jdbc(bigURL, table, connectionProperties);

            } else if (params.getSavemodel() == OracleWriterEnum.OVER_WRITER.getIndexs()) {

                df.write().mode(SaveMode.Overwrite).jdbc(bigURL, table, connectionProperties);

            } else if (params.getSavemodel() == OracleWriterEnum.ERROR_IF_EXIT.getIndexs()) {

                df.write().mode(SaveMode.ErrorIfExists).jdbc(bigURL, table, connectionProperties);

            } else {

                new Exception("Oracle数据库输出算子存储模式错误!");
            }
        }catch (Exception e){

            LOG.error("Oracle数据库输出算子: getMessage: "+e.getMessage());
            LOG.error("Oracle数据库输出算子: getCause: "+e.getCause());

            if (e.getMessage().contains("logon denied")){
                throw new Exception("Oracle数据库输出算子用户名或密码错误！");
            }
            else if(e.getMessage().contains("The Network Adapter could not establish the connection")){
                throw new Exception("Oracle数据库输出算子数据库地址错误！");
            }
            else if(e.getMessage().contains("does not exist")){
                throw new Exception("Oracle数据库输出算子数据库或数据表不存在！");
            }
            else if(e.getMessage().contains("already exists")){
                throw new Exception("Oracle数据库输出算子数据库或数据表已经存在！");
            }
            else if(e.getCause() != null){
                if(e.getCause() instanceof BatchUpdateException) {
                    throw new Exception("Oracle数据库输出算子插入的记录大小超出数据库限制！");
                }

            }else if(e.getMessage().contains("存储模式错误")){

                throw new Exception("Oracle数据库输出算子存储模式错误!");

            }
            throw  new Exception("Oracle数据库输出算子保存数据失败!");
        }



    }
}